import os
os.environ["platform"] = "openi"

from openmind_hub import create_repo,OM_HUB_CACHE,http_get,om_hub_url,upload_file,upload_folder,snapshot_download,om_hub_download,\
                         CommitOperationAdd,create_commit,build_om_headers,try_to_load_from_cache
import time,unittest
from datetime import date
import shutil

def search_tmp_File(pathname):
    matchedFile =[]
    for root,dirs,files in os.walk(pathname):
        for file in files:
            if file.startswith("tmp"):
                fname = os.path.abspath(os.path.join(root, file))
                matchedFile.append(fname)
        return matchedFile
def search_dir_File(pathname,reponame):
    folders = [f for f in os.listdir(pathname) if os.path.isdir(os.path.join(pathname, f))]
    filtered_folders = [f for f in folders if 'openmind' in f or 'model' in f or 'space' in f or 'dataset' in f or reponame in f]
    return filtered_folders

repo_id = "guoxiaozhen/model"+date.today().strftime('%Y%m%d')
reponame = "model"+date.today().strftime('%Y%m%d')
license = "mit"
token = '123456789()'

class TestModel(unittest.TestCase):
    def test_01_openi_create_modelrepo_success(self):
        try:
            create_repo(
                repo_id=repo_id,
                private=False,
                desc="描述内容",
                license=license,
                token=token,
                repo_type="dataset",
                exist_ok=False,
                fullname=None
            )
        except:
            raise Exception("创建仓库有报错")
    def test_02_openi_create_modelrepo_duplicate_existok(self):
        try:
            create_repo(
                repo_id=repo_id,
                private=False,
                desc="描述内容",
                license=license,
                token=token,
                repo_type="model",
                exist_ok=True,
                fullname=None
            )
        except:
            raise Exception("exist_ok=True未生效")
    def test_03_openi_create_modelrepo_duplicate_existnotok(self):
        try:
            create_repo(
                repo_id=repo_id,
                private=False,
                desc="描述内容",
                license=license,
                token=token,
                repo_type="model",
                exist_ok=False,
                fullname=None
            )
            raise Exception("exist_ok=False未生效")
        except:
            pass
    def test_04_openi_create_modelrepo_wrong_token(self):
        try:
            create_repo(
                repo_id=repo_id,
                private=False,
                desc="描述内容",
                license=license,
                token=token + "a",
                repo_type="model",
                exist_ok=True,
                fullname=None
            )
            raise Exception("token未校验")
        except:
            pass
    def test_05_openi_create_modelrepo_wrong_license(self):
        try:
            create_repo(
                repo_id=repo_id,
                private=False,
                desc="描述内容",
                license=license + "2",
                token=token,
                repo_type="model",
                exist_ok=True,
                fullname=None
            )
            raise Exception("license未校验")
        except:
            pass
    def test_06_openi_model_upload_folder_success(self):
        try:
            upload_folder(
                repo_id=repo_id,
                folder_path="/openmind-hub/auto_test/autotest/test_openmind_hub/filedata",
                commit_message="upload folder",
                commit_description=None,
                token=token,
                revision=None,
                # allow_patterns=["*.bin", "*.py"],
                ignore_patterns="*.log",
                # num_threads=5,
            )
        except:
            raise Exception("上传文件报错")
    def test_07_openi_uploadfolder_noexistropo(self):
        try:
            upload_folder(
                repo_id=repo_id+"A",
                folder_path="/openmind-hub/auto_test/autotest/test_openmind_hub/filedata",
                commit_message="upload folder",
                commit_description=None,
                token=token,
                revision=None,
                # allow_patterns=["*.bin", "*.py"],
                ignore_patterns="*.log",
                # num_threads=5,
            )
            raise Exception("上传文件到不存在的仓库未报错")
        except:
            pass
    def test_08_openi_uploadfolder_wrongtoken(self):
        try:
            upload_folder(
                repo_id=repo_id ,
                folder_path="/openmind-hub/auto_test/autotest/test_openmind_hub/filedata",
                commit_message="upload folder",
                commit_description=None,
                token=token + "A",
                revision=None,
                # allow_patterns=["*.bin", "*.py"],
                ignore_patterns="*.log",
                # num_threads=5,
            )
            raise Exception("上传文件未校验token")
        except:
            pass
    def test_09_openi_uploadfolder_duplicate(self):
        try:
            upload_folder(
                repo_id=repo_id,
                folder_path="/openmind-hub/auto_test/autotest/test_openmind_hub/filedata",
                commit_message="upload folder",
                commit_description=None,
                token=token,
                revision=None,
                # allow_patterns=["*.bin", "*.py"],
                ignore_patterns="*.log",
                # num_threads=5,
            )
        except :
            raise Exception("重复上传文件报错")
    # def test_10_openi_dataset_upload_folder_success(self):
    #     try:
    #         upload_folder(
    #             repo_id=repo_id,
    #             folder_path="./filedata",
    #             commit_message="upload folder",
    #             commit_description=None,
    #             token=token,
    #             revision=None,
    #             # allow_patterns=["*.bin", "*.py"],
    #             ignore_patterns="*.log",
    #             num_threads=5,
    #             repo_type="dataset",
    #         )
    #     except:
    #         raise Exception("上传文件报错")
    def test_11_openi_create_commit_success(self):
        try:
            operations = [CommitOperationAdd(path_or_fileobj=r"/openmind-hub/auto_test/autotest/test_openmind_hub/requirements.txt", path_in_repo="1.txt")]
            create_commit(
                repo_id=repo_id,
                operations=operations,
                commit_message="create commit",
                commit_description=None,
                token=token,
                revision="main",
                num_threads=5,
            )
        except:
            raise Exception("上传文件报错")
    def test_12_openi_create_commit_wrongtoken(self):
        try:
            operations = [CommitOperationAdd(
                path_or_fileobj=r"/openmind-hub/auto_test/autotest/test_openmind_hub/requirements.txt",
                path_in_repo="1.txt")]
            create_commit(
                repo_id=repo_id,
                operations=operations,
                commit_message="create commit",
                commit_description=None,
                token=token+"a",
                revision="main",
                num_threads=5,
            )
            raise Exception("创建提交，错误的token未报错")
        except :
            pass
    def test_13_openi_create_commit_noexistrepo(self):
        try:
            operations = [CommitOperationAdd(
                path_or_fileobj=r"/openmind-hub/auto_test/autotest/test_openmind_hub/requirements.txt",
                path_in_repo="1.txt")]
            create_commit(
                repo_id=repo_id+"Sd",
                operations=operations,
                commit_message="create commit",
                commit_description=None,
                token=token,
                revision="main",
                num_threads=5,
            )
            raise Exception("创建提交，不存在的仓库未报错")
        except :
            pass
    def test_14_openi_omhubdownload_success(self):
        try:
            om_hub_download(
                repo_id=repo_id,
                filename="app.py",
                subfolder=None,
                user_agent=None,
                local_dir="/openmind-hub/auto_test/autotest/test_openmind_hub/filedata",
                force_download=False,
                token=token,
                local_files_only=False,
                revision="v1"
            )
            if os.path.exists("/openmind-hub/auto_test/autotest/test_openmind_hub/filedata/app(1).py"):
                print("文件存在。")
            else:
                raise AssertionError("文件不存在。")
        except:
            raise Exception("下载文件有报错")
        else:
            os.remove("/openmind-hub/auto_test/autotest/test_openmind_hub/filedata/app(1).py")
    # def test_15_openi_om_hub_download_wrongtoken(self):
    #     try:
    #         om_hub_download(
    #             repo_id=repo_id,
    #             filename="app.py",
    #             subfolder=None,
    #             user_agent=None,
    #             local_dir="./filedata",
    #             force_download=False,
    #             token=token+"s",
    #             local_files_only=False,
    #             revision="v1"
    #         )
    #         raise Exception("omhubdownload未校验token")
    #     except:
    #        pass
    def test_16_openi_om_hub_download_noexistrepo(self):
        try:
            om_hub_download(
                repo_id=repo_id+"A",
                filename="app.py",
                subfolder=None,
                user_agent=None,
                local_dir="/openmind-hub/auto_test/autotest/test_openmind_hub/filedata",
                force_download=False,
                token=token,
                local_files_only=False,
                revision="v1"
            )
            raise Exception("omhubdownload未校验repo")
        except:
           pass
    def test_17_openi_snapshot_download_success(self):
        try:
            snapshot_download(
                repo_id=repo_id,
                repo_type=None,
                revision=None,
                # cache_dir=OM_HUB_CACHE,
                local_dir_use_symlinks="auto",
                library_name=None,
                library_version=None,
                user_agent=None,
                proxies=None,
                resume_download=True,
                force_download=False,
                token=token,
                local_files_only=False,
                # allow_patterns="allowed_folder\*",    # 只下载allowed_folder目录下的文件
                # ignore_patterns=["*.log", "*.txt"],    # 不下载以.log和.txt结尾的文件
                max_workers=8,
                tqdm_class=None,
                endpoint=None,
            )
            time.sleep(3)
            dirs=search_dir_File("/openmind-hub/",reponame)
            self.assertTrue(len(dirs),"下载的文件不存在")
        except:
           raise Exception("下载文件有报错")
        else:
            shutil.rmtree("/openmind-hub/"+reponame)
    def test_18_openi_snapshot_download_noexist_repo(self):
        try:
            snapshot_download(
                repo_id=repo_id+"a",
                repo_type=None,
                revision=None,
                # cache_dir=OM_HUB_CACHE,
                #local_dir="space_test0516",
                local_dir_use_symlinks="auto",
                library_name=None,
                library_version=None,
                user_agent=None,
                proxies=None,
                resume_download=True,
                force_download=False,
                token=token,
                local_files_only=False,
                # allow_patterns="allowed_folder\*",    # 只下载allowed_folder目录下的文件
                # ignore_patterns=["*.log", "*.txt"],    # 不下载以.log和.txt结尾的文件
                max_workers=8,
                tqdm_class=None,
                endpoint=None,
            )
            raise Exception("下载文件未校验repo")
        except :
           pass
    def test_19_openi_httpget_public_omhuburl_success(self):
        try:
            http_get(
                url=om_hub_url(repo_id=repo_id, filename="1.txt"),
                temp_file=open('/openmind-hub/auto_test/autotest/test_openmind_hub/filedata/test1.txt', 'wb'),
                # displayed_filename="2.txt",
                proxies=None,
                resume_size=0,
                headers=None,
                expected_size=None,
                _nb_retries=5,
            )
            time.sleep(3)
            with open('/openmind-hub/auto_test/autotest/test_openmind_hub/filedata/test1.txt', 'r', encoding='utf-8') as f:
                f_content = f.read()
            self.assertEqual(f_content,"gradio")
        except:
           raise Exception("下载文件有报错")
        else:
            f_content = f_content.replace("gradio", "test")
            with open('/openmind-hub/auto_test/autotest/test_openmind_hub/filedata/test1.txt', 'w', encoding='utf-8') as f:
                f.write(f_content)
    def test_20_openi_httpget_public_noexistrepo(self):
        try:
            http_get(
                url=om_hub_url(repo_id=repo_id+"a", filename="1.txt"),
                temp_file=open('/openmind-hub/auto_test/autotest/test_openmind_hub/filedata/app.py', 'wb'),
                # displayed_filename="2.txt",
                proxies=None,
                resume_size=0,
                headers=None,
                expected_size=None,
                _nb_retries=5,
            )
            raise Exception("未校验repoid")
        except :
           pass
    def test_21_openi_httpget_public_noexistfile(self):
        try:
            http_get(
                url=om_hub_url(repo_id=repo_id , filename="invalid.txt"),
                temp_file=open('/openmind-hub/auto_test/autotest/test_openmind_hub/filedata/app.py', 'wb'),
                # displayed_filename="2.txt",
                proxies=None,
                resume_size=0,
                headers=None,
                expected_size=None,
                _nb_retries=5,
            )
            raise Exception("未校验文件是否存在")
        except:
            pass
    def test_22_openi_httpget_private_success(self):
        repo_id = "guoxiaozhen/modelprivate"+date.today().strftime('%Y%m%d')
        try:
            create_repo(
                repo_id=repo_id,
                private=True,
                desc="描述内容",
                license=license,
                token=token,
                repo_type="model",
                exist_ok=True,
                fullname=None
            )
            operations = [CommitOperationAdd(
                path_or_fileobj=r"/openmind-hub/auto_test/autotest/test_openmind_hub/requirements.txt",
                path_in_repo="1.txt")]
            create_commit(
                repo_id=repo_id,
                operations=operations,
                commit_message="create commit",
                commit_description=None,
                token=token,
                revision="main",
                num_threads=5,
            )
            http_get(
                url=om_hub_url(repo_id=repo_id, filename="1.txt",token=token),
                temp_file=open('/openmind-hub/auto_test/autotest/test_openmind_hub/filedata/test1.txt', 'wb'),
                # displayed_filename="2.txt",
                proxies=None,
                resume_size=0,
                headers=build_om_headers(token=token),
                expected_size=None,
                _nb_retries=5,
            )
            time.sleep(3)
            with open('/openmind-hub/auto_test/autotest/test_openmind_hub/filedata/test1.txt', 'r', encoding='utf-8') as f:
                f_content = f.read()
            self.assertEqual(f_content, "gradio")
        except:
           raise Exception("下载文件有报错")
    def test_23_openi_httpget_private_notoken(self):
        repo_id = "guoxiaozhen/modelprivate"+date.today().strftime('%Y%m%d')
        try:
            http_get(
                url=om_hub_url(repo_id=repo_id, filename="1.txt", token=token),
                temp_file=open('/openmind-hub/auto_test/autotest/test_openmind_hub/filedata/test1.txt', 'wb'),
                # displayed_filename="2.txt",
                proxies=None,
                resume_size=0,
                headers=None,
                expected_size=None,
                _nb_retries=5,
            )
            raise Exception("未校验token")
        except:
           pass
    def test_24_openi_omhuburl_private_notoken(self):
        repo_id = "guoxiaozhen/modelprivate"+date.today().strftime('%Y%m%d')
        try:
            http_get(
                url=om_hub_url(repo_id=repo_id, filename="1.txt"),
                temp_file=open('/openmind-hub/auto_test/autotest/test_openmind_hub/filedata/test1.txt', 'wb'),
                # displayed_filename="2.txt",
                proxies=None,
                resume_size=0,
                headers=build_om_headers(token=token),
                expected_size=None,
                _nb_retries=5,
            )
            raise Exception("未校验token")
        except:
           pass
    def test_25_openi_try_to_load_from_cache_success(self):
        try:
            a = try_to_load_from_cache(
                repo_id=repo_id,
                filename="/openmind-hub/auto_test/autotest/test_openmind_hub/filedata/test1.txt",
            )
            self.assertIsNotNone(a, "未正确返回文件路径")
        except:
           raise Exception("查找文件在缓存中的路径有报错")
    def test_26_openi_try_to_load_from_cache_noexistfile(self):
        try:
            a = try_to_load_from_cache(
                repo_id=repo_id,
                filename="/openmind-hub/auto_test/autotest/test_openmind_hub/filedata/invalid.txt",
            )
            self.assertIsNone(a, "未正确返回文件路径")
        except:
           raise Exception("查找文件在缓存中的路径有报错")


if __name__ == '__main__':
    unittest.main()
